package com.orderpay_detect

import com.orderpay_detect.bean.{OrderEvent, OrderResult}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 20:37
  *                 分类支付成功事件以及15分钟订单超时事件
  */
object OrderTimeout {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val filePath: String = "OrderPayDetect/src/main/resources/OrderLog.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val socketStream: DataStream[String] = env.socketTextStream("192.168.88.106", 7777)

    val dataStream: DataStream[OrderEvent] = fileStream.map(data => {
      val arr: Array[String] = data.split(",")
      OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
    }).assignAscendingTimestamps(_.timestamp * 1000L)

    val orderResultStream: DataStream[OrderResult] = dataStream.keyBy(_.orderId)
      .process(new OrderPayMatchResult)

    orderResultStream.print("payed")
    orderResultStream.getSideOutput(new OutputTag[OrderResult]("timeout")).print("timeout")

    env.execute("order timeout without cep")

  }

  //自定义实现KeyedProcessFunction
  class OrderPayMatchResult extends KeyedProcessFunction[Long, OrderEvent, OrderResult] {
    //定义状态，标识符表示create，pay是否已经来过，定时器时间戳
    lazy val isCreatedState: ValueState[Boolean] = getRuntimeContext
      .getState(new ValueStateDescriptor[Boolean]("is-created", classOf[Boolean]))

    lazy val isPayedState: ValueState[Boolean] = getRuntimeContext
      .getState(new ValueStateDescriptor[Boolean]("is-payed", classOf[Boolean]))

    lazy val timerTsState: ValueState[Long] = getRuntimeContext
      .getState(new ValueStateDescriptor[Long]("timer-ts", classOf[Long]))

    //定义侧输出流标签
    val orderTimeoutOutputTag = new OutputTag[OrderResult]("timeout")

    override def processElement(i: OrderEvent, context: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, collector: Collector[OrderResult]): Unit = {
      //先拿到当前状态
      val isPayed = isPayedState.value()
      val isCreated = isCreatedState.value()
      val timerTs = timerTsState.value()
      //判断当前类型，看是create还是pay
      //来的是create，要继续判断是否pay过
      if (i.eventType == "create") {
        //如果已经支付过，正常支付，输出匹配成功的结果
        if (isPayed) {
          collector.collect(OrderResult(i.orderId, "payed successfully"))
          context.timerService().deleteEventTimeTimer(timerTs)
          isCreatedState.clear()
          isPayedState.clear()
          timerTsState.clear()

        } else {
          //如果还没有pay过，注册定时器，等待15分钟
          val ts = i.timestamp * 1000L + 15 * 60 * 1000L
          context.timerService().registerEventTimeTimer(ts)
          //更新状态
          timerTsState.update(ts)
          isCreatedState.update(true)
        }
      }
      //判断当前来的是pay，要判断是否create过
      else if (i.eventType == "pay") {
        if (isCreated) {
          //如果已经create过，匹配成功，还要判断一下pay时间是否超过了定时器时间
          if (i.timestamp * 1000L < timerTs) {
            //没有超时，正常输出
            collector.collect(OrderResult(i.orderId, "payed successfully"))
          } else {
            //已经超时，输出超时
            context.output(orderTimeoutOutputTag, OrderResult(i.orderId, "payed but already timeout"))
          }
          //只要输出结果，当前order处理已经结束，清空状态和定时器
          context.timerService().deleteEventTimeTimer(timerTs)
          isCreatedState.clear()
          isPayedState.clear()
          timerTsState.clear()
        } else {
          //如果create没来，注册定时器，等到pay的时间就可以了
          context.timerService().registerEventTimeTimer(i.timestamp * 1000L)
          //更新状态
          timerTsState.update(i.timestamp * 1000L)
          isPayedState.update(true)

        }
      }

    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
        //定时器触发
      //如果pay来了，没等到create
      if(isPayedState.value()){
        ctx.output(orderTimeoutOutputTag,OrderResult(ctx.getCurrentKey,"payed but not found create log"))
      }else{
        //create来了，没有来pay
        ctx.output(orderTimeoutOutputTag,OrderResult(ctx.getCurrentKey,"order timeout"))
      }
      //清空状态
      isCreatedState.clear()
      isPayedState.clear()
      timerTsState.clear()

    }
  }

}
